package com.shujia.util

import com.alibaba.fastjson.JSON
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

object CarUtil {
  /**
    * 读取过车数据
    *
    */

  def loadKafkaCar(ssc: StreamingContext, group: String): DStream[Car] = {

    val kafkaParams: Map[String, Object] = Map[String, Object](
      "bootstrap.servers" -> "master:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> group,
      "auto.offset.reset" -> "earliest"
    )
    //消费的topic
    val topics = Array("car")

    //创建一个流
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //取出数据
    val lines: DStream[String] = stream.map(record => {
      record.value()
    })


    //将json数据解析出来
    val cars: DStream[Car] = lines.map(line => {
      val car: Car = JSON.parseObject(line, classOf[Car])
      car
    })

    cars

  }

  case class Car(car: String, city_code: String, county_code: String, card: Long, camera_id: String, orientation: String, road_id: Long, time: Long, speed: Double)

}
